/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import jvs.peer.PresenceListener;
import jvs.peer.io.FileTransfer.Monitor;
import jvs.peer.io.PeerStatus.Status;
import jvs.peer.io.Session.Callback;
import jvs.peer.sec.AccessControl;
import jvs.peer.util.Log;
import jvs.peer.util.Net;

/**
 * @author qiangli
 * 
 */
public final class ChannelManager implements Lifecycle {

	private static final int SLEEP = 200;

	private static final int MAX_CHANNELS = 1024;

	private long maxIdle = 3 * 60 * 1000L; //5 * 60 * 1000L;

	private long checkInterval = 30 * 1000L; //60 * 1000L;

	private class Cleaner extends Thread {
		@Override
		public void run() {
			if (Log.DEBUG) {
				Log.log(this, "Starting Channel cleaner...");
			}
			
			while (running) {
				try {
					long now = System.currentTimeMillis();
					
					if (Log.DEBUG) {
						Log.log(this, "channels: " + channels.size() + " now: " + now);
					}
					
					for (Iterator<Map.Entry<Object, Channel>> it = channels
							.entrySet().iterator(); it.hasNext();) {
						Map.Entry<Object, Channel> m = it.next();
						Channel c = m.getValue();
						long elapsed = now - c.timestamp;
						if (elapsed > maxIdle) {
							try {
								
								if (Log.DEBUG) {
									Log.log(this, "closing: " + c);
								}
								
								c.close();
							
							} catch (Exception e) {
								e.printStackTrace();
							}
							it.remove();
							inbound.remove(c);
						}
					}
					Thread.sleep(checkInterval);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
			if (Log.DEBUG) {
				Log.log(this, "Channel cleaner stopped");
			}
			
		}
	}

	private volatile boolean running;

	private Map<Object, Channel> channels = new ConcurrentHashMap<Object, Channel>();

	private BlockingQueue<Channel> inbound = new LinkedBlockingQueue<Channel>(
			MAX_CHANNELS);

	private static long id = System.currentTimeMillis();

	private Lifecycle service;

	private MessageDriver driver;

	private FileTransfer transfer;

	private Messenger messenger;

	private Notifier notifier;

	private PeerStatus peerStatus;

	private Cleaner cleaner;

	private Properties attributes;

	private String jid;

	private int port;

	synchronized static private String nextThread() {
		++id;
		return "id-" + id;
	}

	private static List<ChannelManager> instances = new ArrayList<ChannelManager>();

	private String driverName = null;
	private String driverClass = null;

	public static final String NETWORK_XMPP = "xmpp";
	public static final String NETWORK_BTSPP = "btspp";

	public static final String CHANNEL_DRIVER_CLASS = "channel.driver.class";

	static {
		//default
		instances.add(new ChannelManager(NETWORK_XMPP, "jvs.peer.xmpp.Jabber"));
		instances.add(new ChannelManager(NETWORK_BTSPP, "jvs.peer.bt.Bluetooth"));
	}
	
	public static ChannelManager getInstance() {
		return getInstance(NETWORK_XMPP);
	}
	
	public static ChannelManager getInstance(String name) {
		for (ChannelManager i : instances) {
			if (name.equalsIgnoreCase(i.driverName)) {
				return i;
			}
		}
		ChannelManager man = new ChannelManager(name, null);
		instances.add(man);
		return man;
	}

	private ChannelManager(String realm, String driver) {
		this.driverName = realm;
		this.driverClass = driver;
	}
	
	public String getRealm() {
		return this.driverName;
	}

	public void init(Properties attributes) throws Exception {
		this.attributes = attributes;
		//
		Object obj = createDriver(attributes);

		driver = (MessageDriver) obj;
		transfer = (FileTransfer) obj;
		messenger = (Messenger) obj;
		notifier = (Notifier) obj;
		peerStatus = (PeerStatus) obj;
		//
		service = (Lifecycle) obj;
		service.init(attributes);
		//
		jid = attributes.getProperty("xmpp.jid");
		port = parseInt(attributes.getProperty("http.port"), 10080);
	}

	private int parseInt(String p, int def) {
		try {
			return Integer.parseInt(p);
		} catch (Exception e) {

		}
		return def;
	}

	public void startup() throws Exception {

		cleanup(); //in case shutdown was not called
		
		//
		service.startup();
		//
		MessageDispatcher disp = new MessageDispatcher(this, channels, driver);

		Thread dt = new Thread(disp);
		dt.setDaemon(true);
		dt.start();
		//
		running = true;
		cleaner = new Cleaner();
		cleaner.start();

		//
		for (PresenceListener pl : presenceListeners) {
			notifier.register(pl);
		}
		presenceListeners.clear();
	}

	public void shutdown() throws Exception {
		running = false;
		if (service != null) {
			service.shutdown();
		}
		if (cleaner != null) {
			cleaner.interrupt();
		}

		driver = null;
		transfer = null;
		messenger = null;
		notifier = null;
		peerStatus = null;
		//
		service = null;

		presenceListeners.clear();
		
		//
		cleanup();
	}

	public Channel acceptChannel() throws IOException {
		try {
			return inbound.take();
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new IOException(e.getMessage());
		}
	}

	/**
	 * new server channel
	 * 
	 * @throws IOException
	 */
	public Channel createServerChannel(String tid, String peer)
			throws IOException {
		MessageDriver drv = getDriver(peer);
		
		String id = peer + "|" + tid;
		Channel c = new Channel(id, drv);
		//
		Header h = new Header();
		h.setThread(tid);
		h.setPeer(peer);
		h.setMode(MessageConstant.response);

		c.setHeader(h);
		//
		if (!inbound.offer(c)) {
			throw new IOException("Queue max size exceeded");
		}
		channels.put(id, c);
		if (Log.DEBUG) {
			Log.log(this, "server channel created. tid: " + tid);
		}
		return c;
	}

	/**
	 * new client channel
	 * 
	 * @param peer
	 * @return
	 * @throws IOException
	 */
	public Channel openChannel(String peer) throws IOException {
		MessageDriver drv = getDriver(peer);
		
		String tid = nextThread();
		String id = "|" + tid;
		Channel c = new Channel(id, drv);

		Header h = new Header();
		h.setThread(tid);
		h.setPeer(peer);
		h.setMode(MessageConstant.request);

		c.setHeader(h);

		channels.put(id, c);

		c.open();
		if (Log.DEBUG) {
			Log.log(this, "client channel opened. tid: " + tid);
		}
		return c;
	}
	
	private void cleanup() {
		//clean up channels
		if (Log.DEBUG) {
			Log.log(this, "cleanup channels, totoal: " + channels.size());
		}
		for (Iterator<Map.Entry<Object, Channel>> it = channels
				.entrySet().iterator(); it.hasNext();) {
			Map.Entry<Object, Channel> m = it.next();
			Channel c = m.getValue();
			try {
				c.close();
			} catch (Exception e) {
				if (Log.DEBUG) {
					e.printStackTrace();
				}
			}
			it.remove();
			inbound.remove(c);
		}
		
		channels.clear();
		inbound.clear();
	}

	private MessageDriver getDriver(String peer) throws IOException {
		if (driver == null) {
			throw new IOException("Peer service not ready. " + peer);
		}
		return driver;
	}

	public Messenger getMessenger() {
		return messenger;
	}

	public MessageDriver getDriver() {
		return driver;
	}

	public FileTransfer getTransfer() {
		return transfer;
	}

	public void send(Message msg) throws Exception {
		driver.send(msg);
	}

	public void remove(Channel channel) {
		try {
			channels.remove(channel.getId());
			inbound.remove(channel);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public boolean ping(String peer) {
		return driver.ping(peer);
	}

	protected Object createDriver(Properties attrs) throws Exception {
		final String name = attrs.getProperty(CHANNEL_DRIVER_CLASS);
		Class<?> cls = (Class<?>) Class.forName(name == null ? driverClass
				: name);
		return cls.newInstance();
	}

	private Set<PresenceListener> presenceListeners = new HashSet<PresenceListener>();

	public void register(String peer, String tid, MessageListener ml) {
		notifier.register(peer, tid, ml);
	}

	public void unregister(MessageListener ml) {
		notifier.unregister(ml);
	}

	public void unregister(PresenceListener pl) {
		if (notifier != null) {
			notifier.unregister(pl);
		}
		presenceListeners.remove(pl);
	}

	public void register(PresenceListener pl) {
		if (notifier != null) {
			notifier.register(pl);
		} else {
			// postpone
			presenceListeners.add(pl);
		}
	}

	public void send(String to, String tid, String message) throws Exception {
		messenger.send(to, tid, message);
	}

	public void invite(String peer) throws Exception {
		messenger.invite(peer);
	}

	public Monitor receive(String from, String filename, File saveas)
			throws Exception {
		return transfer.receive(from, filename, saveas);
	}

	public Monitor send(String to, String filename, File file) throws Exception {
		return transfer.send(to, filename, file);
	}

	public void changeStatus(Status status) {
		if (peerStatus != null) {
			peerStatus.changeStatus(status);
		}
	}

	public Status checkStatus(String peer) {
		if (peerStatus == null) {
			return Status.OFF;
		}
		return peerStatus.checkStatus(peer);
	}

	public String[] getNetAddress(String peer) {
		if (peerStatus == null) {
			return null;
		}
		return peerStatus.getNetAddress(peer);
	}

	public Session createSession(String peer, Callback cb) throws Exception {
		checkConnect();
		return messenger.createSession(peer, cb);
	}

	private void checkConnect() {
		if (running == false) {
			throw new ServiceUnavailableException("Service not started");
		}
	}

	public void release() {
		cleanup();
	}

	public AccessControl getAccessControl() {
		if (driver == null || driver.getAccessControl() == null) {
			throw new ServiceUnavailableException("Shell service not started");
		}
		return driver.getAccessControl();
	}

	public boolean isLocal(String peer) {
		if (attributes == null) {
			return false; // we really don't know
		}
		String[] pa = peer.split("/", 2);
		String id = attributes.getProperty("xmpp.jid");
		if (pa.length == 0 || id == null) {
			return false;
		}
		return pa[0].equals(id);
	}

	public String getJid() {
		return jid;
	}

	public int getPort() {
		return port;
	}

	public String getIp() {
		return Net.ip();
	}

	public boolean isAvailable() {
		return (driver != null && driver.available());
	}

	public boolean isAvailable(long wait) {
		try {
			while (!isAvailable() && wait > 0) {
				wait -= SLEEP;
				if (Log.DEBUG) {
					Log.log(this, "waiting for driver... time: " + wait);
				}
				Thread.sleep(SLEEP);
			}
			return isAvailable();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

	public String[] getNetAddress(String peer, long wait) {
		try {
			while (checkStatus(peer) != (Status.AVAIL) && wait > 0) {
				wait -= SLEEP;
				if (Log.DEBUG) {
					Log.log(this, "waiting for peer... " + peer + " time: "
							+ wait);
				}
				Thread.sleep(SLEEP);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		String[] addr = getNetAddress(peer);
		return addr;
	}

	public boolean isAvailable(String peer) {
		return (peerStatus != null && peerStatus.checkStatus(peer) == PeerStatus.Status.AVAIL);
	}

	public void removePeer(String peer) {
		getAccessControl().removeUser(peer);
	}

	public void addPeer(String peer) {
		getAccessControl().addUser(peer, peer);
	}
}
